package com.dlink.sqljobsync.service;

import com.dlink.sqljobsync.autoconfig.SqlJobAutoConfig;
import com.dlink.sqljobsync.event.KafkaNoticeEvent;
import com.dlink.sqljobsync.event.MysqlNoticeEvent;
import com.dlink.sqljobsync.propertie.DlinkSqlProperties;
import com.dlink.sqljobsync.utils.DlinkSyncHttpTool;
import com.dlink.sqljobsync.utils.Utility;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.event.ApplicationEventMulticaster;
import org.springframework.stereotype.Component;
import org.springframework.util.ResourceUtils;

import javax.annotation.PostConstruct;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author huangjun
 */
@Component
@Slf4j
public class DlinkSqlSyncFaceService  {

    @PostConstruct
    public void initTool(){
        DlinkSyncHttpTool.setDlinkSqlSyncFaceService(this);
    }

    @Autowired
    private Utility utility;

    @Autowired
    private DlinkSqlProperties dlinkSqlProperties;

    @Autowired
    private ApplicationEventMulticaster eventMulticaster;

    public String executeSql() {
        /**自动装配需要开启 */
        if(dlinkSqlProperties.isEnable() == true){
            Properties prop = null;
            try {
                prop = utility.loadProperties();
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(0);
            }
            //获取当前系统运行时间
            long nowTimestamp = System.currentTimeMillis();
            if(prop != null){
                long lastTimestamp_date_stat = Long.valueOf(prop.getProperty("lastTimestamp_date_stat"));
                if(lastTimestamp_date_stat < nowTimestamp){
                    //正常运行,记录上次的运行时间必须在当前时间之前
                    if(!dlinkSqlProperties.isStatus()){
                        //正常执行不需要进行替换的SQL脚本目录
                        String sqlFilePath = prop.getProperty("file_path");
                        List<String> fileNames = new ArrayList<String>();
                        File file = null;
                        //获取当前应用路径
                        String javafxRuntimePath = null;
                        //获取classpath下默认sql目录进行加载所有sql脚本
                        try {
                            String classpathUrlPrefix = ResourceUtils.CLASSPATH_URL_PREFIX;
                            classpathUrlPrefix = System.getProperty(classpathUrlPrefix);
                            log.info(classpathUrlPrefix);
                            file = ResourceUtils.getFile("classpath:" +sqlFilePath);
                        } catch (FileNotFoundException e) {
                            javafxRuntimePath = System.getProperty("user.dir");
                            file = new File(javafxRuntimePath, sqlFilePath);
                        }
                        utility.findFileList(file, fileNames);
                        if(fileNames.size() == 0){
                            return "当前目录： "  + sqlFilePath + "没有运行的sql脚本，请准备脚本后再运行。";
                        }
                        execJobSql(fileNames, prop);
                    }else{
                        //执行需要替换参数的SQL脚本
                    }
                }else{
                    return "运行失败，任务不允许并行执行。。。。";
                }
            }
        }
        return null;
    }

    /**
     * 获取SQL目录下的脚本进行批量执行
     * @param fileNames 文件名
     * @param prop 执行脚本SQL类型配置类型
     */
    private void execJobSql(List<String> fileNames, Properties prop){
        StringBuffer buffer = null;
        ApplicationEvent event = null;
        //读取执行Job Sql脚本参数
        String flinkConnectorType = prop.getProperty("flink_connector_type");
        String flinkysType = prop.getProperty("flinkys_type");
        String flinkAddress = prop.getProperty("flink_address");
        for(String sqlFile: fileNames){
            buffer = utility.getReadFileTxt(sqlFile);
            if(buffer != null){
                switch (flinkConnectorType)
                {
                    case "KafKa" :
                        event = new KafkaNoticeEvent(buffer.toString(), flinkysType, flinkAddress);
                        break;
                    case "Mysql" :
                        event = new MysqlNoticeEvent(buffer.toString(), flinkysType, flinkAddress);
                    break;
                    default:
                        break;
                }
                SqlJobAutoConfig.noticeEvent = event;
                eventMulticaster.multicastEvent(SqlJobAutoConfig.noticeEvent);
            }
        }
    }

}
